#include <stdio.h>
#include <string.h>
#include "Define.h"
#include "Package.h"
#include "MQTTNode.h"
#include "Auth/Auth.h"
#include "MsgHandler.h"
#include "MQTT/Params.h"
#include "MqttValidor.h"
#include "Core/Logger.h"
#include "MQTT/MQTTError.h"
#include "Core/Interface.h"

static int32_t mqttConnectSign(MQTTEntity* entity);
static void notifyMqttEvent(MQTTEntity *entity, AiotEventType type);
static void reclaimUserPropertiesMem(MQTTEntity *entity, UserProperty **userPropertyList);
static int32_t compareMqttTopic(char *topic, uint32_t topicLen, char *dstTopic, uint32_t dstTopicLen);


int32_t handlePingRequest(MQTTEntity *entity) {
    return sendPingResp(entity);
}


int32_t handlePingResponse(MQTTEntity *entity, uint8_t *input, uint32_t len) {

    RecvMQTTMessage packet;
    uint64_t rtt = getSystemInterface()->getTimestamp() - entity->aiotEntity.hblInfo.lastSendAt;

    if (len != 0) {
        return mqttRecvInvalidPongPacket;
    }

    /** heartbreat rtt stats */
    if (rtt < 10000) {
        entity->aiotEntity.netStats.rtt = (entity->aiotEntity.netStats.rtt + rtt) / 2;
    }

    memset(&packet, 0, sizeof(RecvMQTTMessage));

    if (entity->recvHandler != nullptr) {
        packet.type = recvHeartBeatRes;
        entity->recvHandler((void *)entity, &packet, entity->aiotEntity.context);
    }

    return aiotSuccess;
}

int32_t handlerPublish(MQTTEntity *entity, uint8_t *input, uint32_t len, uint8_t qos, uint32_t remainLen) {

    uint8_t subFound = 0,offset = 0,useAliasTopic  = 0;
    uint32_t idx = 0,topicLen = 0,propertyLen = 0,totalPropLen = 0;
    uint16_t packetId = 0,utf8StrLen = 0;

    RecvMQTTMessage packet;
    void *context = nullptr;
    MQTTSubNode *subNode = nullptr;
    MQTTTopicAliasNode topicAliasNode = {0};

    MQTTSubHandlerNode *handlerNode = nullptr;
    ListHead handlerListCopy;
    PubProperty pubProp = {0};
    
    if (input == nullptr || len == 0 || qos > MQTT_QOS1) {
        return mqttRecvInvalidPublishPacket;
    }

    memset(&packet, 0, sizeof(RecvMQTTMessage));

    packet.data.publish.qos = qos;
    packet.type = recvPublish;

    utf8StrLen = input[idx++] << 8;
    utf8StrLen |= input[idx++];

    uint8_t topic[utf8StrLen+1];
    memset(topic, 0, sizeof(topic));
    memcpy(topic, &input[idx], utf8StrLen);
    packet.data.publish.topic = topic;
    packet.data.publish.topicLen = utf8StrLen;
    idx += utf8StrLen;

    /** Packet Id For QOS1 */
    if (len < idx) {
        return mqttRecvInvalidPublishPacket;
    }

    if (qos == MQTT_QOS1) {
        if (len < idx + 2) {
            return mqttRecvInvalidPublishPacket;
        }

        packetId = input[idx++] << 8;
        packetId |= input[idx++];

    }

    int32_t res = parsePublishProperties(entity, &input[idx], &pubProp, &propertyLen, &offset, remainLen);
    if (res != aiotSuccess) {
        return res;
    }

    totalPropLen = propertyLen + offset ;
    idx += totalPropLen;

    packet.data.publish.pubProp = &pubProp;
    useAliasTopic = handleTopicAlias(entity, &pubProp, packet.data.publish.topicLen,packet.data.publish.topic, &topicAliasNode);

    /** Payload Len */
    if ((int64_t)len - MQTT_PUBLISH_TOPICLEN_LEN - (int64_t)packet.data.publish.topicLen < 0) {
        return mqttRecvInvalidPublishPacket;
    }

    packet.data.publish.payloadLen = len - MQTT_PUBLISH_TOPICLEN_LEN - packet.data.publish.topicLen - totalPropLen;
    if (qos == MQTT_QOS1) {
        packet.data.publish.payloadLen -= MQTT_PACKETID_LEN;
    }

    uint8_t payload[packet.data.publish.payloadLen+1];
    memset(payload, 0, sizeof(payload));
    memcpy(payload, &input[idx], packet.data.publish.payloadLen);
    packet.data.publish.payload = payload;

    /** Publish Ack For QOS1 */
    if (qos == MQTT_QOS1) {
        sendPublicAck(entity, packetId);
    }

    /** User previos topic alias */
    if (1 == useAliasTopic) {
        packet.data.publish.topic = topicAliasNode.topic;
        packet.data.publish.topicLen = strlen(packet.data.publish.topic);
    }

    /** debug */
    topicLen = (uint32_t)packet.data.publish.topicLen;

    loggerOut(aiotTrace,__FILE__,__LINE__,"publish: %u.%s\n",topicLen,packet.data.publish.topic);

    /** search Packet Handler In sublist */
    initialAiotListHead(&handlerListCopy);
    getSystemInterface()->enterMutex(entity->subMutex);
    AIOT_LIST_FOR_EACH_ENTRY(subNode, &entity->subList, linkedNode, MQTTSubNode) {
        if (compareMqttTopic(subNode->topic, (uint32_t) (strlen(subNode->topic)), packet.data.publish.topic,
                             packet.data.publish.topicLen) == aiotSuccess) {
            appendMqttHandlerList(entity, &handlerListCopy, &subNode->handleList, &subFound);
        }
    }

    getSystemInterface()->leaveMutex(entity->subMutex);

    AIOT_LIST_FOR_EACH_ENTRY(handlerNode, &handlerListCopy,linkedNode, MQTTSubHandlerNode) {

        context = (handlerNode->context == nullptr) ? (entity->aiotEntity.context) : (handlerNode->context);
        handlerNode->handler(entity, &packet, context);
    }

    destroyMqttSubHandlerList(entity, &handlerListCopy);

    /** user Data Default Packet Handler*/
    if (entity->recvHandler && subFound == 0) {
        entity->recvHandler((void *)entity, &packet, entity->aiotEntity.context);
    }

    /* Reclaims Mem of User Properties */
    reclaimUserPropertiesMem(entity, pubProp.userProperty);
    return aiotSuccess;
}

int32_t handlerPublishAck(MQTTEntity *entity, uint8_t *input, uint32_t len) {

    RecvMQTTMessage packet;

    /** even maltform puback received, flow control should be re-calculated as well */
    mqttFlowControlInc(entity);

    if (input == nullptr) {
        return mqttRecvInvalidPubAckPacket;
    }

    if (len < 2) {
        return mqttRecvInvalidPubAckPacket;
    }

    memset(&packet, 0, sizeof(RecvMQTTMessage));
    packet.type = recvPublishAck;

    /** packet id*/
    packet.data.publishAck.packetId = input[0] << 8;
    packet.data.publishAck.packetId |= input[1];

    /** remove packet from rePublist */
    getSystemInterface()->enterMutex(entity->pubMutex);
    removeMqttPubList(entity, packet.data.publishAck.packetId);
    getSystemInterface()->leaveMutex(entity->pubMutex);

    /** user ctrl packet handler */
    if (entity->recvHandler) {
        entity->recvHandler((void *)entity, &packet, entity->aiotEntity.context);
    }

    return aiotSuccess;
}

void handlerSubUnsubAck(MQTTEntity *entity, uint8_t *input, uint32_t len, uint8_t packetType) {

    RecvMQTTMessage packet;
    int32_t res = aiotSuccess;
    uint32_t idx = 0;
    uint32_t propertyLen = 0;
    uint8_t offset = 0;

    if (input == nullptr || len == 0) {
        return;
    }

    /** Packet ID */
    memset(&packet, 0, sizeof(RecvMQTTMessage));
    if (packetType == MQTT_SUBACK_PKT_TYPE) {
        packet.type = recvSubscribeAck;
        packet.data.subscribeAck.packetId = input[idx] << 8;
        packet.data.subscribeAck.packetId |= input[idx + 1];
        getSystemInterface()->enterMutex(entity->subMutex);
        updateMqttSubList(entity, nullptr, packet.data.subscribeAck.packetId, NULL, 0, 1, 1);
        getSystemInterface()->leaveMutex(entity->subMutex);
    } else {
        packet.type = recvUnSubscribeAck;
        packet.data.unsubscribeAck.packetId = input[idx] << 8;
        packet.data.unsubscribeAck.packetId |= input[idx + 1];
    }

    idx += 2;

    /** Properties */
    res = readVariableByteInterger(&(input[idx]), &propertyLen, &offset);
    if (res != aiotSuccess) {
        return;
    }

    idx = idx + offset + propertyLen;

    /** Suback Flags */
    if (packetType == MQTT_SUBACK_PKT_TYPE) {
        if (input[idx] == MQTT_SUBACK_RCODE_MAXQOS0 ||
            input[idx] == MQTT_SUBACK_RCODE_MAXQOS1 ||
            input[idx] == MQTT_SUBACK_RCODE_MAXQOS2) {

            packet.data.subscribeAck.res = aiotSuccess;
            packet.data.subscribeAck.maxQos = input[idx];
        } else if (input[idx] == MQTT_SUBACK_RCODE_FAILURE) {
            packet.data.subscribeAck.res = mqttSubAckFailed;
        } else {
            packet.data.subscribeAck.res = mqttSubAckUnknown;
        }
    }

    if (entity->recvHandler) {
        entity->recvHandler((void *)entity, &packet, entity->aiotEntity.context);
    }
}

int32_t handlerDisconnect(MQTTEntity *entity, uint8_t *input, uint32_t len) {

    RecvMQTTMessage packet;
    uint8_t reasonCode;
    if (input == nullptr) {
        return mqttRecvInvalidDisconnectPacket;
    }

    if (len < 2) {
        return mqttRecvInvalidDisconnectPacket;
    }

    memset(&packet, 0, sizeof(RecvMQTTMessage));
    packet.type = recvDisconnect;

    reasonCode = input[0];
    packet.data.disconnect.reasonCode = reasonCode;

    /** User Ctrl Packet Handler */
    if (entity->recvHandler) {
        entity->recvHandler((void *)entity, &packet, entity->aiotEntity.context);
    }

    /** close socket connect with mqtt broker */
    destroyAiotNet(&entity->aiotEntity);
    notifyDisconnectEvent(entity, networkDisconnect);

    loggerOut(aiotDebug,__FILE__,__LINE__,"MQTT receives server disconnect, disconnect\n");
    return aiotSuccess;
}

int32_t handleConnectAck(MQTTEntity *entity, uint8_t *connAck, uint32_t remainLen) {

    int32_t  res = aiotSuccess;
    int32_t  ret = aiotSuccess;

    ConnAckProperty connackProp = {0};
    RecvMQTTMessage packet = {0};

    if (connAck[0] != 0x00) {
        return mqttConnectAckParseError;
    }

    /** First Byte is Conack Flag */
    if (connAck[1] == MQTT_CONNACK_RCODE_ACCEPTED) {
        res = aiotSuccess;
    } else if (connAck[1] == MQTT_V5_CONNACK_RCODE_UNACCEPTABLE_PROTOCOL_VERSION) {
        loggerOut(aiotDebug,__FILE__,__LINE__,"invalid protocol version, disconeect\n");
        res = mqttVersionUnsupport;
    } else if (connAck[1] == MQTT_V5_CONNACK_RCODE_SERVER_UNAVAILABLE) {
        loggerOut(aiotDebug,__FILE__,__LINE__,"server unavailable, disconnect\n");
        res = mqttServerUnavalidable;
    } else if (connAck[1] == MQTT_V5_CONNACK_RCODE_BAD_USERNAME_PASSWORD) {
        loggerOut(aiotDebug,__FILE__,__LINE__,"bad username or password, disconnect\n");
        res = mqttAccountInvalid;
    } else if (connAck[1] == MQTT_V5_CONNACK_RCODE_NOT_AUTHORIZED) {
        loggerOut(aiotDebug,__FILE__,__LINE__,"authorize fail, disconnect\n");
        res = mqttNotAuthorized;
    } else {
        loggerOut(aiotDebug,__FILE__,__LINE__,"unknown connect ack:%d, disconnect\n", connAck[1]);
        res = mqttConnectAckUnknown;
    }
    if(res!=aiotSuccess) {
        return res;
    }

    /** Second Byte is Reason Code */
    packet.data.connectAck.reasonCode = connAck[1];

    /** Properties Starts from 3rd Byte */
    ret = parseConAckProperties(entity, &connAck[2], &connackProp, remainLen);
    if (ret != aiotSuccess) {
        return ret;
    }

    packet.data.connectAck.property = connackProp;
    packet.type = recvConnectAck;

    if (entity->recvHandler) {
        entity->recvHandler((void *)entity, &packet,entity->aiotEntity.context);
    }

    if (nullptr != connackProp.clientId) {
        getSystemInterface()->memFree(connackProp.clientId);
    }
    /** Reclaims Mem of User Properties */
    reclaimUserPropertiesMem(entity, connackProp.userProperty);

    return res;
}

void notifyDisconnectEvent(MQTTEntity *entity, AiotDisConnectEvent disconnect) {

    if (entity->hasConnected != 1 || entity->disconnected != 0) {
        return;
    }

    AiotEvent event;
    entity->disconnected = 1;
    memset(&event, 0, sizeof(AiotEvent));
    event.type = aiotDisconnect;
    event.data.disconnect = disconnect;

    if (entity->aiotEntity.eventHandler) {
        entity->aiotEntity.eventHandler((void *)entity, &event,entity->aiotEntity.context);
    }

    getSystemInterface()->enterMutex(entity->processMutex);
    notifyProcessHandler(entity, &event, nullptr);
    getSystemInterface()->leaveMutex(entity->processMutex);

    /** 重置list所有节点 */
    getSystemInterface()->enterMutex(entity->subMutex);
    resetAllMqttSubListStatus(entity);
    getSystemInterface()->leaveMutex(entity->subMutex);

    /**断开连接*/
    destroyAiotNet(&entity->aiotEntity);
}

void  notifyConnectEvent(MQTTEntity *entity) {

    entity->disconnected = 0;
    entity->aiotEntity.hblInfo.lostTimes = 0;
    if (entity->hasConnected == 0) {
        entity->hasConnected = 1;
        notifyMqttEvent(entity, aiotConnect);
    } else {
        notifyMqttEvent(entity, aiotReConnect);
    }
}

void handleProcessData(MQTTEntity *entity, MQTTEvent *event) {

    MQTTProcessDataNode *node = nullptr;

    getSystemInterface()->enterMutex(entity->processMutex);
    AIOT_LIST_FOR_EACH_ENTRY(node, &entity->processList,linkedNode, MQTTProcessDataNode) {
        node->processData.handler(node->processData.context, nullptr, event);
    }

    getSystemInterface()->leaveMutex(entity->processMutex);
}

uint8_t handleTopicAlias(MQTTEntity *entity, PubProperty *pubProp, uint32_t topicLen,char *topic, MQTTTopicAliasNode *aliasNode) {

    MQTTTopicAliasNode *topicAliasNode = nullptr;
    uint8_t useAliasTopic = 0;
    if (nullptr != pubProp && pubProp->topicAlias > 0) {
        if (topicLen > 0) {
            /** A topic appears first time during this connection. */
            MQTTBuffer aliasTopic = {.buffer = (uint8_t *)topic,.len = topicLen,};
            /** Insert it into topic alias list */
            insertMqttTopicAliasList(entity, &aliasTopic, pubProp->topicAlias,&(entity->rxTopicAliasList));

        } else {

            /** topic len is 0. User previos topic alias */
            getSystemInterface()->enterMutex(entity->topicAliasMutex);
            AIOT_LIST_FOR_EACH_ENTRY(topicAliasNode, &entity->rxTopicAliasList, linkedNode,MQTTTopicAliasNode) {

                if (topicAliasNode->topicAlias == pubProp->topicAlias) {
                    loggerOut(aiotDebug,__FILE__,__LINE__,"found a rx topic alias\n");
                    useAliasTopic = 1;
                    memcpy(aliasNode, topicAliasNode, sizeof(MQTTTopicAliasNode));
                    break;
                }
            }

            getSystemInterface()->leaveMutex(entity->topicAliasMutex);
        }
    }

    return useAliasTopic;
}

void notifyProcessHandler(MQTTEntity *entity, AiotEvent *event, MQTTEvent *mqttEvent) {

    MQTTProcessDataNode *node = nullptr;

    AIOT_LIST_FOR_EACH_ENTRY(node, &entity->processList,linkedNode, MQTTProcessDataNode) {
        node->processData.handler(node->processData.context, event, mqttEvent);
    }
}

int32_t handleMqttConnect(MQTTEntity *entity, ConnectProperty *prop) {

    int32_t res = 0, readLen = 0, sendLen = 0;
    SocketType socketType = tcpClient;
    uint8_t *connPkt = nullptr, *connackBuffer = nullptr;
    uint8_t connackFixedHeader = 0;
    uint32_t connPktLen = 0, remainLen = 0;

    AiotServerAddress *aiotServer = &entity->aiotEntity.aiotServer;
    if (aiotServer->host == nullptr) {
        return paramsMissingHost;
    }

    res = mqttConnectSign(entity);
    if (res != aiotSuccess) {
        return res;
    }

    res = createAiotNet(&entity->aiotEntity, socketType);
    if (res != aiotSuccess) {
        return res;
    }

    /** Remove All Topic Alias Relationship */
    remoteTopicAliasList(entity);

    /** network stats */
    entity->aiotEntity.netStats.connectAt = getSystemInterface()->getTimestamp();
    res = getSystemInterface()->netEstablish(entity->aiotEntity.netHandle);
    if (res != aiotSuccess) {
        destroyAiotNetUnlock(&entity->aiotEntity);
        entity->aiotEntity.netStats.failedAt = entity->aiotEntity.netStats.connectAt;
        entity->aiotEntity.netStats.failedCode = res;
        return networkEstablishFailed;
    }

    entity->aiotEntity.netStats.connectUsed =
            getSystemInterface()->getTimestamp() - entity->aiotEntity.netStats.connectAt;

    /** Get MQTT Connect Packet */
    res = mqttConnectPkt(entity, &connPkt, &connPktLen, prop);
    if (res != aiotSuccess) {
        return res;
    }

    /** Send MQTT Connect Packet */
    res = writeMqttData(entity, connPkt, connPktLen, &sendLen, entity->aiotEntity.sendDataTimeout);
    getSystemInterface()->memFree(connPkt);
    if (res != aiotSuccess) {
        if (res == networkWriteLessData) {
            loggerOut(aiotTrace, __FILE__, __LINE__, "[CONNECT]mqtt connect packet send timeout: %d\n",
                      entity->aiotEntity.sendDataTimeout);
        } else {
            loggerOut(aiotError, __FILE__, __LINE__, "[CONNECT]mqtt connect: send connect pkt errno %d!\n", res);
            destroyAiotNetUnlock(&entity->aiotEntity);
        }

        return res;
    }

    /** Receive MQTT Connect ACK Fixed Header Byte */
    res = readMqttMessage(entity, &connackFixedHeader, MQTT_FIXED_HEADER_LEN, &readLen,
                          entity->aiotEntity.recvDataTimeout);
    if (res != aiotSuccess) {
        if (res == networkReadLessData) {
            loggerOut(aiotTrace, __FILE__, __LINE__, "[CONNECT]MQTT connack packet recv fixed header timeout: %u\n",
                      entity->aiotEntity.recvDataTimeout);
        } else {
            loggerOut(aiotError, __FILE__, __LINE__, "[CONNECT]mqtt connect: recv connect ack pkt errno %d!\n", res);
            destroyAiotNetUnlock(&entity->aiotEntity);
        }

        return res;
    }

    if (connackFixedHeader != MQTT_CONNACK_PKT_TYPE) {
        return mqttConnectAckParseError;
    }

    /** Receive MQTT Connect ACK remain len */
    res = readRemainlen(entity, &remainLen);
    if (res < aiotSuccess) {
        if (res == networkReadLessData) {
            loggerOut(aiotTrace, __FILE__, __LINE__, "[CONNECT]MQTT connack packet recv remain length timeout: %d\n",
                      entity->aiotEntity.recvDataTimeout);
        }

        return res;
    }

    connackBuffer = getSystemInterface()->memMalloc(remainLen);
    if (nullptr == connackBuffer) {
        return memoryMallocFailed;
    }

    memset(connackBuffer, 0, remainLen);

    /** Receive MQTT Connect ACK Variable Header and Properties */
    res = readMqttMessage(entity, connackBuffer, remainLen, &readLen, entity->aiotEntity.recvDataTimeout);
    if (res != aiotSuccess) {
        if (res == networkReadLessData) {
            loggerOut(aiotTrace, __FILE__, __LINE__,
                      "[CONNECT]MQTT connack packet variable header and property recv timeout: %d\n",
                      entity->aiotEntity.recvDataTimeout);
        }

        getSystemInterface()->memFree(connackBuffer);
        return res;
    }

    res = handleConnectAck(entity, connackBuffer, remainLen);
    getSystemInterface()->memFree(connackBuffer);
    if (res != aiotSuccess) {
        loggerOut(aiotError, __FILE__, __LINE__, "[CONNECT]mqtt connect: handle connect pkt errno %d!\n", res);
        destroyAiotNetUnlock(&entity->aiotEntity);
        return res;
    }

    return mqttConnectSuccess;
}

int32_t sendPingResp(MQTTEntity *entity) {
    int32_t res = 0,sendLen = 0;
    uint8_t pkt[2] = {0};

    /** Packet Type */
    pkt[0] = MQTT_PINGRESP_PKT_TYPE;

    /** Remaining Lengh */
    pkt[1] = 0;

    getSystemInterface()->enterMutex(entity->aiotEntity.sendMutex);
    res = writeMqttData(entity, pkt, 2,&sendLen, entity->aiotEntity.sendDataTimeout);
    getSystemInterface()->leaveMutex(entity->aiotEntity.sendMutex);
    if (res < aiotSuccess) {
        if (res != networkWriteLessData) {
            loggerOut(aiotError,__FILE__,__LINE__,"send ping response: send data errno %d!\n",res);
            destroyAiotNet(&entity->aiotEntity);
        }

        return res;
    }

    return aiotSuccess;
}

int32_t sendPublicAck(MQTTEntity *entity, uint16_t packetId) {

    int32_t res = 0,sendLen = 0;
    uint8_t pkt[4] = {0};

    /** Packet Type */
    pkt[0] = MQTT_PUBACK_PKT_TYPE;

    /** Remaining Lengh */
    pkt[1] = 2;

    /* Packet Id */
    pkt[2] = (uint16_t)((packetId >> 8) & 0x00FF);
    pkt[3] = (uint16_t)((packetId) & 0x00FF);

    getSystemInterface()->enterMutex(entity->aiotEntity.sendMutex);
    res = writeMqttData(entity, pkt, 4,&sendLen, entity->aiotEntity.sendDataTimeout);
    getSystemInterface()->leaveMutex(entity->aiotEntity.sendMutex);
    if (res < aiotSuccess) {
        if (res != networkWriteLessData) {
            loggerOut(aiotError,__FILE__,__LINE__,"send publish ack: send data errno %d!\n",res);
            destroyAiotNet(&entity->aiotEntity);
        }

        return res;
    }

    return aiotSuccess;
}

int32_t readRemainbytes(MQTTEntity *entity, uint32_t remainlen, uint8_t **output) {

    int32_t res = 0,readLen = 0;
    uint8_t *remain = nullptr;

    if (remainlen > 0) {
        remain = getSystemInterface()->memMalloc(remainlen);
        if (remain == nullptr) {
            return memoryMallocFailed;
        }

        memset(remain, 0, remainlen);

        res = readMqttMessage(entity, remain, remainlen,&readLen, entity->aiotEntity.recvDataTimeout);
        if (res != aiotSuccess) {
            getSystemInterface()->memFree(remain);
            if (res == networkReadLessData) {
                return malformedRemainingBytes;
            } else {
                return res;
            }
        }
    }

    *output = remain;
    return aiotSuccess;
}

int32_t reSendPublish(MQTTEntity *entity) {

    int32_t res = 0,sendLen = 0;
    uint64_t time_now = 0;
    MQTTPubNode *node = nullptr;

    getSystemInterface()->enterMutex(entity->pubMutex);
    AIOT_LIST_FOR_EACH_ENTRY(node, &entity->pubList, linkedNode, MQTTPubNode) {
        time_now = getSystemInterface()->getTimestamp();
        if (time_now < node->sendAt) {
            node->sendAt = time_now;
        }

        if ((time_now - node->sendAt) >= entity->rePublishTimeout) {
            node->sendAt = time_now;
            getSystemInterface()->enterMutex(entity->aiotEntity.sendMutex);
            res = writeMqttData(entity, node->packet, node->length,&sendLen,entity->aiotEntity.sendDataTimeout);
            getSystemInterface()->leaveMutex(entity->aiotEntity.sendMutex);
            loggerOut(aiotDebug,__FILE__,__LINE__,"republish: packetId %d send data res: %d!\n",node->packetId, res);
            if (res != aiotSuccess) {
                getSystemInterface()->leaveMutex(entity->pubMutex);
                if (res != networkWriteLessData) {
                    loggerOut(aiotError,__FILE__,__LINE__,"republish: packetId %d send errno %d!\n",node->packetId,res);
                    destroyAiotNet(&entity->aiotEntity);
                }

                return res;
            }
        }
    }

    getSystemInterface()->leaveMutex(entity->pubMutex);
    return aiotSuccess;
}

int32_t reSendSubscribe(MQTTEntity *entity) {

    int32_t res = 0,sendLen = 0;
    uint64_t time_now = 0;
    MQTTSubNode *node = nullptr;

    getSystemInterface()->enterMutex(entity->subMutex);
    AIOT_LIST_FOR_EACH_ENTRY(node, &entity->subList, linkedNode, MQTTSubNode) {
        time_now = getSystemInterface()->getTimestamp();
        if (time_now < node->sendAt) {
            node->sendAt = time_now;
        }

        //未收到SubscribeAck且超过订阅时间的需要重发
        if (((time_now - node->sendAt) >= entity->reSubscribeTimeout) && node->status==0) {
            node->sendAt = time_now;
            getSystemInterface()->enterMutex(entity->aiotEntity.sendMutex);
            res = writeMqttData(entity, node->packet, node->length,&sendLen,entity->aiotEntity.sendDataTimeout);
            getSystemInterface()->leaveMutex(entity->aiotEntity.sendMutex);

            if (res != aiotSuccess) {
                getSystemInterface()->leaveMutex(entity->subMutex);
                if (res != networkWriteLessData) {
                    loggerOut(aiotError,__FILE__,__LINE__,"[resubscribe]%s: send  errno %d!\n",node->topic, res);
                    destroyAiotNet(&entity->aiotEntity);
                }

                return res;
            }
        }
    }

    getSystemInterface()->leaveMutex(entity->subMutex);
    return aiotSuccess;
}


int32_t sendHeartbeat(MQTTEntity *entity) {

    int32_t res = aiotSuccess,sendLen = 0;
    uint8_t pingPkt[2] = { MQTT_PINGREQ_PKT_TYPE, 0x00 };
    getSystemInterface()->enterMutex(entity->aiotEntity.sendMutex);
    res = writeMqttData(entity, pingPkt, 2,&sendLen, entity->aiotEntity.sendDataTimeout);
    getSystemInterface()->leaveMutex(entity->aiotEntity.sendMutex);
    if (res != aiotSuccess) {
        if (res != networkWriteLessData) {
            loggerOut(aiotError,__FILE__,__LINE__,"send heartbeat: send  errno %d!\n",res);
            destroyAiotNet(&entity->aiotEntity);
        }

        return res;
    }

    return aiotSuccess;
}

int32_t sendReconnect(MQTTEntity *entity) {

    int32_t res = networkClosed;
    uint64_t current = 0;
    AiotReconnectInfo* reconnectInfo = &entity->aiotEntity.reconnectInfo;
    uint32_t interval = reconnectInfo->interval;
    if (reconnectInfo->backOffEnabled){
        interval = reconnectInfo->interval* (reconnectInfo->counter + 1) + reconnectInfo->randMs;
    }

    if (entity->aiotEntity.netHandle != nullptr) {
        return aiotSuccess;
    }

    getSystemInterface()->enterMutex(entity->aiotEntity.dataMutex);
    getSystemInterface()->enterMutex(entity->aiotEntity.sendMutex);
    getSystemInterface()->enterMutex(entity->aiotEntity.recvMutex);
    current = getSystemInterface()->getTimestamp();
    if (current < reconnectInfo->lastRetryTime) {
        reconnectInfo->lastRetryTime = current;
    }

    if (current >= (reconnectInfo->lastRetryTime + interval)) {

        loggerOut(aiotDebug,__FILE__,__LINE__,"mqtt network disconnect, try to reconnecting...\n");
        res = handleMqttConnect(entity, nullptr);
        reconnectInfo->lastRetryTime = getSystemInterface()->getTimestamp();
        if (reconnectInfo->backOffEnabled) {

            if(mqttConnectSuccess == res){
                reconnectInfo->counter = 0;
            } else if (reconnectInfo->counter < 60) {
                reconnectInfo->counter++;
            }
        }
    }

    getSystemInterface()->leaveMutex(entity->aiotEntity.recvMutex);
    getSystemInterface()->leaveMutex(entity->aiotEntity.sendMutex);
    getSystemInterface()->leaveMutex(entity->aiotEntity.dataMutex);
    return res;
}

int32_t readMqttMessage(MQTTEntity *entity, uint8_t *buffer, uint32_t len, int32_t *readLen, uint32_t timeout) {

    int32_t res = aiotSuccess;
    if (entity->aiotEntity.netHandle == nullptr) {
        return networkClosed;
    }

    res = getSystemInterface()->netRecv(entity->aiotEntity.netHandle, buffer, len,readLen,timeout,nullptr);
    if (res != aiotSuccess) {
        res = networkRecvError;
    } else if (*readLen != len) {
        res = networkReadLessData;
    }

    return res;
}

int32_t readRemainlen(MQTTEntity *entity, uint32_t *remainlen) {

    int32_t res = 0,readLen = 0;
    uint8_t ch = 0;
    uint32_t multiplier = 1;
    uint32_t mqttRemainLen = 0;

    do {

        res = readMqttMessage(entity, &ch, 1,&readLen, entity->aiotEntity.recvDataTimeout);
        if (res != aiotSuccess) {
            return res;
        }

        mqttRemainLen += (ch & 127) * multiplier;
        if (multiplier > 128 * 128 * 128) {
            return mqttMalformedRemainingLen;
        }

        multiplier *= 128;
    } while ((ch & 128) != 0);

    *remainlen = mqttRemainLen;
    return aiotSuccess;
}

int32_t writeMqttData(MQTTEntity *entity, uint8_t *buffer, uint32_t len, int32_t *sendLen, uint32_t timeout) {

    int32_t res = aiotSuccess;
    if (entity->aiotEntity.netHandle == nullptr) {
        res = networkClosed;
    }

    res = getSystemInterface()->netSend(entity->aiotEntity.netHandle,buffer,len,sendLen,timeout,nullptr);
    if (res != aiotSuccess) {
        res = networkSendError;
    } else if (*sendLen != len) {
        res = networkWriteLessData;
    }

    return res;
}

int32_t sendSubUnsub(MQTTEntity *entity, char *topic, uint16_t topicLen, uint8_t qos,uint8_t pktType, GeneralProperty *property) {

    int32_t res = 0,sendLen = 0;
    uint16_t packetId = 0;
    uint8_t *pkt = nullptr;
    uint32_t idx = 0, pktLen = 0, remainLen = 0;
    uint32_t propertyTotalLen = 0;
    uint32_t propertyLen = 0, propertyLenOffset = 0;
    uint8_t propertyLenArray[4] = {0};

    if (nullptr != property) {
        res = mqttGeneralPropertyValid(property);
        if (res != aiotSuccess) {
            return res;
        }

        propertyLen = getGeneralPropertyLen(property);
    }

    /** Property Len */
    encodeRemainLen(propertyLen, &propertyLenArray[0], &propertyLenOffset);
    propertyTotalLen = propertyLenOffset + propertyLen;

    remainLen = MQTT_PACKETID_LEN + MQTT_UTF8_STR_EXTRA_LEN + topicLen + propertyTotalLen;
    if (pktType == MQTT_SUB_PKT_TYPE) {
        remainLen += MQTT_REQUEST_QOS_LEN;
    }

    pktLen = MQTT_FIXED_HEADER_LEN + MQTT_REMAINLEN_MAXLEN + remainLen;

    pkt = getSystemInterface()->memMalloc(pktLen);
    if (pkt == nullptr) {
        return memoryMallocFailed;
    }

    memset(pkt, 0, pktLen);

    /** Subscribe/Unsubscribe Packet Type */
    if (pktType == MQTT_SUB_PKT_TYPE) {
        pkt[idx++] = MQTT_SUB_PKT_TYPE | MQTT_SUB_PKT_RESERVE;
    } else if (pktType == MQTT_UNSUB_PKT_TYPE) {
        uint8_t unsubFlag = MQTT_UNSUB_PKT_TYPE | MQTT_UNSUB_PKT_RESERVE;
        pkt[idx++] = unsubFlag;
    }

    /** Remaining Length */
    encodeRemainLen(remainLen, &pkt[idx], &idx);

    /** Packet Id */
    packetId = getMqttPacketId(entity);
    pkt[idx++] = (uint8_t)((packetId >> 8) & 0x00FF);
    pkt[idx++] = (uint8_t)((packetId) & 0x00FF);
    if (pktType == MQTT_SUB_PKT_TYPE) {
        loggerOut(aiotTrace,__FILE__,__LINE__,"[SUBSCRIBE]mqtt topic: %s, packetId:%d\n",topic, packetId);
    } else if(pktType == MQTT_UNSUB_PKT_TYPE) {
        loggerOut(aiotTrace,__FILE__,__LINE__,"[UNSUBSCRIBE]mqtt topic: %s, packetId:%d\n",topic, packetId);
    }

    writeGeneralProp(pkt, &idx, property, propertyLenArray, propertyLenOffset);

    /** Topic */
    pkt[idx++] = (uint8_t)((topicLen >> 8) & 0x00FF);
    pkt[idx++] = (uint8_t)((topicLen) & 0x00FF);
    memcpy(&pkt[idx], topic, topicLen);
    idx += topicLen;

    /** QOS */
    if (pktType == MQTT_SUB_PKT_TYPE) {
        pkt[idx++] = qos;
    }

    pktLen = idx;

    if (pktType == MQTT_SUB_PKT_TYPE) {
        /** 更新list节点 */
        MQTTBuffer topicBuffer = {0};
        topicBuffer.buffer = topic;
        topicBuffer.len = topicLen;
        getSystemInterface()->enterMutex(entity->subMutex);
        updateMqttSubList(entity, &topicBuffer, packetId, pkt, pktLen, 0, 0);
        getSystemInterface()->leaveMutex(entity->subMutex);
    }

    getSystemInterface()->enterMutex(entity->aiotEntity.sendMutex);
    res = writeMqttData(entity, pkt, pktLen,&sendLen,entity->aiotEntity.sendDataTimeout);
    getSystemInterface()->leaveMutex(entity->aiotEntity.sendMutex);
    if (res != aiotSuccess) {
        getSystemInterface()->memFree(pkt);
        if (res != networkWriteLessData) {
            loggerOut(aiotError,__FILE__,__LINE__,"send sub and unsub: send errno %d!\n",res);
            destroyAiotNet(&entity->aiotEntity);
        }

        return res;
    }

    getSystemInterface()->memFree(pkt);
    return res;
}

int32_t processTxTopicAlias(MQTTEntity *entity, MQTTBuffer *topic, uint16_t *aliasIdPtr) {

    MQTTTopicAliasNode *topicAliasNode = nullptr;
    uint16_t aliasId = 0;
    int32_t replaceTopicWithAlias = 0;

    /** Topic Alias */
    getSystemInterface()->enterMutex(entity->topicAliasMutex);
    AIOT_LIST_FOR_EACH_ENTRY(topicAliasNode, &entity->txTopicAliasList,linkedNode,MQTTTopicAliasNode) {

        if (memcmp(topicAliasNode->topic, topic->buffer, topic->len) == 0) {
            aliasId = topicAliasNode->topicAlias;
            replaceTopicWithAlias = 1;
            break;
        }
    }

    getSystemInterface()->leaveMutex(entity->topicAliasMutex);
    /** save the tx topic in tx_topic_alias_list */
    if (0 == replaceTopicWithAlias) {
        aliasId = entity->txTopicAlias;
        aliasId++;
        /** make sure that it does not exceed tx_topic_alias_max */
        if (aliasId <= entity->txTopicAliasSize) {
            entity->txTopicAlias = aliasId;
            getSystemInterface()->enterMutex(entity->topicAliasMutex);
            insertMqttTopicAliasList(entity, topic, aliasId, &(entity->txTopicAliasList));
            getSystemInterface()->leaveMutex(entity->topicAliasMutex);
        } else {
            /** exceed tx_topic_alias_max, no more alias could be used */
            aliasId = 0;
        }
    }

    *aliasIdPtr = aliasId;
    return replaceTopicWithAlias;
}

uint8_t sendSubResult(MQTTEntity *entity) {
    MQTTSubNode *node = NULL;
    AIOT_LIST_FOR_EACH_ENTRY(node, &entity->subList, linkedNode, MQTTSubNode) {
        if(node->status == 0) {
            return 0;
        }
    }

    return 1;
}

static void notifyMqttEvent(MQTTEntity *entity, AiotEventType type) {

    AiotEvent event;
    memset(&event, 0, sizeof(AiotEvent));
    event.type = type;

    if (entity->aiotEntity.eventHandler) {
        entity->aiotEntity.eventHandler((void *)entity, &event, entity->aiotEntity.context);
    }

    getSystemInterface()->enterMutex(entity->processMutex);
    notifyProcessHandler(entity, &event, nullptr);
    getSystemInterface()->leaveMutex(entity->processMutex);
}

static void reclaimUserPropertiesMem(MQTTEntity *entity, UserProperty **userPropertyList) {

    int iter = 0;
    for (iter = 0; iter < MAX_USER_PROPERTY; iter++) {
        UserProperty *prop = userPropertyList[iter];
        if (nullptr != prop) {
            getSystemInterface()->memFree(prop);
            userPropertyList[iter] = nullptr;
        }
    }
}

static int32_t compareMqttTopic(char *topic, uint32_t topicLen, char *dstTopic, uint32_t dstTopicLen) {

    uint32_t idx = 0, cmpIdx = 0;

    for (idx = 0, cmpIdx = 0; idx < topicLen; idx++) {
        /** compare topic alreay out of bounds */
        if (cmpIdx >= dstTopicLen) {
            /** compare success only in case of the left string of topic is "/#" */
            if ((topicLen - idx == 2) && (memcmp(&topic[idx], "/#", 2) == 0)) {
                return aiotSuccess;
            } else {
                return mqttTopicCompareFailed;
            }
        }

        /** if topic reach the '#', compare success */
        if (topic[idx] == '#') {
            return aiotSuccess;
        }

        if (topic[idx] == '+') {
            /** wildcard + exist */
            for (; cmpIdx < dstTopicLen; cmpIdx++) {
                if (dstTopic[cmpIdx] == '/') {
                    /** if topic already reach the bound, compare topic should not contain '/' */
                    if (idx + 1 == topicLen) {
                        return mqttTopicCompareFailed;
                    } else {
                        break;
                    }
                }
            }
        } else {
            /** compare each character */
            if (topic[idx] != dstTopic[cmpIdx]) {
                return mqttTopicCompareFailed;
            }

            cmpIdx++;
        }
    }

    /** compare topic should be reach the end */
    if (cmpIdx < dstTopicLen) {
        return mqttTopicCompareFailed;
    }

    return aiotSuccess;
}


static int32_t mqttConnectSign(MQTTEntity* entity) {

    int32_t res = aiotSuccess;
//    if (entity->username != nullptr && entity->password != nullptr && entity->clientId != nullptr) {
//        return res;
//    }

    res = checkAuthInfo(entity);
    if (res != aiotSuccess) {
        return res;
    }

    cleanMqttSign(entity);
    res = authUsername(&entity->username,&entity->aiotEntity);
    if (res != aiotSuccess) {
        cleanMqttSign(entity);
        return res;
    }

    res = authPassword(&entity->password, &entity->aiotEntity);
    if (res != aiotSuccess) {
        cleanMqttSign(entity);
        return res;
    }

    res = authClientId(&entity->clientId, &entity->aiotEntity,entity->extendClientId);
    if (res != aiotSuccess) {
        cleanMqttSign(entity);
        return res;
    }
    loggerOut(aiotTrace,__FILE__,__LINE__,"clientId:%s\n",entity->clientId);
    loggerOut(aiotTrace,__FILE__,__LINE__,"username:%s\n",entity->username);
    loggerOut(aiotTrace,__FILE__,__LINE__,"password:%s\n",entity->password);
    return res;
}